package com.zx.mes.coolmq.hyl.listener;

import com.rabbitmq.client.Channel;
import com.zx.mes.coolmq.hyl.util.MqConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;

/**
 * RabbitMQ 抽象消息监听，所有消息消费者必须继承此类
 * @author hyl
 * @date 2018-3-20
 * @version 1.0.0
 */
public abstract class AbstractMessageListener implements ChannelAwareMessageListener{

    private static final Logger logger = LoggerFactory.getLogger(AbstractMessageListener.class);
    public static final long REDIS_INCREMENT_DELTA = 1;
    public static final String HYL = "-hyl";


    private Jackson2JsonMessageConverter jsonMessageConverter=new Jackson2JsonMessageConverter();


    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 接收消息，子类必须实现该方法
     * @param message           消息对象
     * @param messageConverter  消息转换器
     */
    public abstract void receiveMessage(Message message, MessageConverter messageConverter);

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        MessageProperties messageProperties = message.getMessageProperties();
        Long deliveryTag = messageProperties.getDeliveryTag();

//        Long consumerCount = stringRedisTemplate.opsForValue().increment(messageProperties.getMessageId(), REDIS_INCREMENT_DELTA);
//        Long consumerCount = redisTemplate.opsForHash().increment(MqConstants.MQ_CONSUMER_RETRY_COUNT_KEY,
//                messageProperties.getMessageId(), REDIS_INCREMENT_DELTA);
        Long consumerCount=0L;
        String messageIdCount=messageProperties.getMessageId() + HYL;
        if (stringRedisTemplate.hasKey(messageIdCount)) {
            consumerCount=consumerCount+1;
            stringRedisTemplate.opsForValue().set(messageIdCount, String.valueOf(consumerCount));
        } else {
            stringRedisTemplate.opsForValue().set(messageIdCount, String.valueOf(0));
        }
        consumerCount = Long.valueOf(stringRedisTemplate.opsForValue().get(messageIdCount))+1;
        logger.info("当前消息ID:{} 消费次数:{}",messageProperties.getMessageId(),consumerCount);

//        执行业务，根据执行情况进行消息的ack
        try {
            receiveMessage(message,jsonMessageConverter);
//            成功的回执
            channel.basicAck(deliveryTag,false);

            stringRedisTemplate.delete(messageIdCount);

        } catch (Exception e) {
            logger.error("RabbitMQ 消息消费失败，e:{}",e.getMessage());
            if (consumerCount >= MqConstants.MAX_CONSUMER_COUNT) {
//                入死信队列
                channel.basicReject(deliveryTag,false );
            }else {
//                重回队列，重新消费，按照2的指数级递增
                Thread.sleep((long) ((Math.pow(MqConstants.BASE_NUM,consumerCount))*1000));
                channel.basicNack(deliveryTag,false,true);
            }
            return;
        }

        try {
//            redisTemplate.opsForHash().delete(MqConstants.MQ_CONSUMER_RETRY_COUNT_KEY,
//                    messageProperties.getMessageId());
            stringRedisTemplate.delete(messageProperties.getMessageId());
        } catch (Exception e) {
            logger.error("消息监听redis删除消费异常,e:{}",e);
        }
    }
}
